package com.aotain.nyx.port;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.flink.BasicTuple;

/**
 * 
     *@Title:
     *@Description:
     *@Author:Turk
     *@Since:2017年6月30日
     *@Version:1.1.0
 */
public class PortReduce implements ReduceFunction<Tuple2<String, BasicTuple>>{

	/**
	  *
	  */
	    
	private static final long serialVersionUID = -5752076046421202785L;

	@Override
	public Tuple2<String, BasicTuple> reduce(Tuple2<String, BasicTuple> value1,
			Tuple2<String, BasicTuple> value2) throws Exception {
		BasicTuple AandRet = value1.f1;
		BasicTuple B = value2.f1;
		
		AandRet.setUpStreamOctets(AandRet.getUpStreamOctets() + B.getUpStreamOctets());
		AandRet.setUpStreamPacket(AandRet.getUpStreamPacket() + B.getUpStreamPacket());
		AandRet.setSessionNum(AandRet.getSessionNum() + B.getSessionNum());
		Tuple2<String, BasicTuple> ret = new Tuple2<String, BasicTuple>();
		ret.setFields(value1.f0, AandRet);
		
		return ret;
		    
	}

}

    